package com.chris.core.socket.netty.util;

import com.chris.core.socket.netty.comm.ByteUtil;
import com.chris.core.socket.netty.dto.*;
import com.chris.core.socket.netty.dto.protocol.CmdObjectProtocol;
import com.chris.core.socket.netty.dto.protocol.CustomMessageProtocol;
import com.chris.core.socket.netty.dto.protocol.FileMessageProtocol;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 综合操作基类
 * 苏斌 2022-3-29
 */
public class NettyRunTimeUtil {

    //#region 基础环境变量
    /**
     * 全局-SocketChannel - 集合 (线程安全)
     */
    public static ConcurrentHashMap<String, SocketChannelMapDTO> SocketChannelMap = new ConcurrentHashMap<>();
    /**
     * 全局channels集合
     */
    //public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 运行时-消息断言
     * 对象初始化时候，全局赋值
     */
    public static String runTimeBreakMark = "\r\n";

    /**
     * 上传文件命令
     */
    public final static String _cmdUpFile = "!@#$upfile%^&*";

    /**
     * Attribute数据结构
     */
    public static final AttributeKey<NettyAttributeMapDTO> NETTY_CHANNEL_KEY = AttributeKey.valueOf("netty.channel");

    /**
     * 客服端在线数
     */
    private static int onlineCount = 0;

    /**
     * 上传文件夹
     */
    public static String uploadFilePath = "./uploads";

    //#endregion

    //#region 线程同步 及其他

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        onlineCount--;
    }


    //#endregion

    //#region 发送消息公用


    /**
     * 发送文件（file->FileMessageProtocol)
     *
     * @param file
     * @param channel
     */
    @Deprecated
    public static void sendFileMsg(File file, Channel channel) {
        if (file.isFile() && file.exists()) {
            String uuid = UUID.randomUUID().toString().replace("-", "");

            FileMessageProtocol fmp = new FileMessageProtocol();
            //----------第一个发送，开始命令---》传输基本信息
            fmp.setFileName(file.getName());
            fmp.setFileSize(file.length());
            fmp.setTagA("begin");
            fmp.setTagB(uuid);
            sendObjMsg(fmp,channel);
            try {
                //此处效果是留给接受端创建文件的时间
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //-----------第二个发送，传输数据------------
            //统计开始时间
            long beginTime = System.currentTimeMillis();
            RandomAccessFile randomAccessFile = null;
            try {
                randomAccessFile = new RandomAccessFile(file, "r");
                int length = 0;
                Long start = 0L;
                //实际操作，每个write 操作，要转换 json ,写入+转换 30M~40M 数据，大概是 66ms 每次耗时， 如果 循环 100次，就是 6秒
                //要综合控制，每次 读取内容的长度，太短，会造成循环次数过多，实际影响写入速度
                    /* 缓存控制量 压力测试评估
                    实测 25M 文件
                    1 * 1024b  10385ms
                    2  - 5209
                    3 - 3506
                    4 - 2754
                    5 -2124
                    6-1565
                    7
                    8 1397
                    9 1189
                    10 1078  --》25M/1S
                    15 734
                    20 558   --》50M/1S
                    30 424
                    40 322   --> 77M/1S
                    300  76  --> 328/1S
                    400 73
                    综合评估 1024 * 10 ， 大概 25M/1S 实际写入速度
                     */
                byte[] bytes = new byte[1024 * 10];
                // List<FileDataDTO>  list = new ArrayList<>();
                while ((length = randomAccessFile.read(bytes)) != -1) {
                    long upTime = System.currentTimeMillis();
                    fmp = new FileMessageProtocol();
                    fmp.setFileName(file.getName());
                    fmp.setFileSize(file.length());
                    fmp.setBeginPos(start);
                    fmp.setEndPos(Long.parseLong(length + ""));
                    fmp.setTagA("tranfer");
                    fmp.setTagB(uuid);

                    byte[] bf = new byte[length];
                    System.arraycopy(bytes, 0, bf, 0, length);
                    fmp.setFileData(bf);
                    //新nio方法，内存反转，0拷贝
                    //byteBuffer.flip();
                    if (length < 10240)//最后一个次发送，延迟1秒
                        Thread.sleep(1000);
                    sendObjMsg(fmp,channel);
                    //限流
                    flowControl(length, upTime, 500);
                    // list.add(fddto);
                    start += length;
                }
/*
                for (int i=list.size()-1;i>=0;i--) {
                    channelFuture.channel().writeAndFlush(list.get(i));
                }*/
            } catch (Exception ex) {
                ex.printStackTrace();
            } finally {
                if (randomAccessFile != null) {
                    try {
                        randomAccessFile.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
            //----------第三个发送，结束命令---------------
            fmp = new FileMessageProtocol();
            fmp.setFileName(file.getName());
            fmp.setTagA("end");
            fmp.setTagB(uuid);
            sendObjMsg(fmp,channel);
            //log.debug("======== 文件推送成功 耗时:" + (System.currentTimeMillis() - beginTime) / 1000 + "秒========");
        }
    }

    /**
     * 发送文件（file->FileMessageProtocol-->CustomMessageProtocol)
     * @param file
     * @param channel
     */
    public static void sendCustomFileMsg(File file, Channel channel, Logger log) throws Exception {
        if (file.isFile() && file.exists()) {
            String uuid = UUID.randomUUID().toString().replace("-", "");

            FileMessageProtocol fmp = new FileMessageProtocol();
            //----------第一个发送，开始命令---》传输基本信息
            fmp.setFileName(file.getName());
            fmp.setFileSize(file.length());
            fmp.setTagA("begin");
            fmp.setTagB(uuid);
            //send-----FileMessageProtocol-->CustomMessageProtocol
            CustomMessageProtocol begin_dto = new CustomMessageProtocol();
            begin_dto.setContentFileObject(fmp);
            sendObjMsg(begin_dto,channel);
            try {
                //此处效果是留给接受端创建文件的时间
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //-----------第二个发送，传输数据------------
            //统计开始时间
            long beginTime = System.currentTimeMillis();
            RandomAccessFile randomAccessFile = null;
            try {
                randomAccessFile = new RandomAccessFile(file, "r");
                int length = 0;
                Long start = 0L;
                //实际操作，每个write 操作，要转换 json ,写入+转换 30M~40M 数据，大概是 66ms 每次耗时， 如果 循环 100次，就是 6秒
                //要综合控制，每次 读取内容的长度，太短，会造成循环次数过多，实际影响写入速度
                    /* 缓存控制量 压力测试评估
                    实测 25M 文件
                    1 * 10240  10385ms
                    2  - 5209
                    3 - 3506
                    4 - 2754
                    5 -2124
                    6-1565
                    7
                    8 1397
                    9 1189
                    10 1078  --》25M/1S
                    15 734
                    20 558   --》50M/1S
                    30 424
                    40 322   --> 77M/1S
                    300  76  --> 328/1S
                    400 73
                    综合评估 1024 * 10 ， 大概 25M/1S 实际速度
                     */
                int maxReadLength=51200;// 1024 * 30;
                byte[] bytes = new byte[maxReadLength];//10KB
                while ((length = randomAccessFile.read(bytes)) != -1) {
                    long upTime = System.currentTimeMillis();
                    fmp = new FileMessageProtocol();
                    fmp.setFileName(file.getName());
                    fmp.setFileSize(file.length());
                    fmp.setBeginPos(start);
                    fmp.setEndPos(Long.parseLong(length + ""));
                    fmp.setTagA("tranfer");
                    fmp.setTagB(uuid);

                    byte[] bf = new byte[length];
                    System.arraycopy(bytes, 0, bf, 0, length);
                    fmp.setFileData(bf);
                    //新nio方法，内存反转，0拷贝
                    //byteBuffer.flip();
                    if (length < maxReadLength)//最后一个次发送，延迟0.5秒
                        Thread.sleep(500);

                    //send-----FileMessageProtocol-->CustomMessageProtocol
                    CustomMessageProtocol mdi_dto = new CustomMessageProtocol();
                    mdi_dto.setContentFileObject(fmp);
                    sendObjMsg(mdi_dto,channel);

                    //限流->  每次[5K] 发送量 -> 10240 b/ms = 10240kb/s = 10m/s(实际7.79m)  , 5120 = 5m/s(实际3.94m/s) ,20480 = 20m/s(实际20.17m/s) , 30720 = 30m/s(实际38.11)
                    flowControl(length, upTime,  10240);
                    // list.add(fddto);
                    start += length;
                }
                log.debug("文件发送完毕，大小："+(file.length()/1024/1024)+"M, 耗时： " +(System.currentTimeMillis()-beginTime)/1000 +" 秒 , 平均 "+(file.length()/1024/1024)/(float)((System.currentTimeMillis()-beginTime)/1000)+"m/s" );
            } catch (Exception ex) {
                ex.printStackTrace();
            } finally {
                if (randomAccessFile != null) {
                    try {
                        randomAccessFile.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
            //----------第三个发送，结束命令---------------
            fmp = new FileMessageProtocol();
            fmp.setFileName(file.getName());
            fmp.setTagA("end");
            fmp.setTagB(uuid);
            CustomMessageProtocol end_dto = new CustomMessageProtocol();
            end_dto.setContentFileObject(fmp);
            sendObjMsg(end_dto,channel);
        }
    }




    /**
     * 发送消息
     */
    public static void sendByteMsg(byte[] bytes, Channel channel, String breakMark) {

        ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes, breakMark.getBytes());
        // byte[] data2=breakMark.getBytes();
        // byte[] data3 = new byte[bytes.length + data2.length];

        // System.arraycopy(bytes, 0, data3, 0, bytes.length);

        // System.arraycopy(data2, 0, data3, bytes.length, data2.length);

        channel.writeAndFlush(byteBuf);

    }

    /**
     * 发送消息
     */
    public static void sendObjMsg(Object ob, Channel channel) {
        channel.writeAndFlush(ob);
    }

    /**
     * 发送消息
     */
    public static void sendStrMsg(String msg, Channel channel, String breakMark) {
        channel.writeAndFlush(msg + breakMark);
    }


    //#endregion

    //#region 流量控制

    /**
     * 流量控制 【时间】段内允许通过的 字节数 <= limitspped,如果超过强制停留在该时间段内
     * 流量控制  (目前策略 接收端+ 发送端  被 限流 ，双方需要约定好 ）
     * 因为是并发异步/同步，所以不能阻止并发发送端速度
     * 只能总限制 单线程速度 , 例如 ： 限制 200KB * 100并发 = 20 M / S  ,1000并发 200M， 1W并发 2000M
     *
     * 写入
     *   netty默认设置的高水位为64KB，低水位为32KB.
     *   netty中使用write方法发送数据时，这个数据其实是写到了一个缓冲区中，并未直接发送给接收方
     *   netty使用ChannelOutboundBuffer封装出站消息的发送，所有的消息都会存储在一个链表中，直到缓冲区被flush方法刷新，netty才会将数据真正发送出去。
     *
     * 读取
     *
     *
     * @param passedlen  每秒读取字节 (b！,非KB）
     * @param beginTime  开始时间(毫秒)
     * @param limitspped 限制速度 b / ms (注：b!非KB/毫秒） ， x毫秒内允许通过字节数 例如： limitspped=200=(200b/ms)=(200KB/s) limitspped=500(500b/ms)=(500KB/s)
     */
    public static void flowControl(long passedlen, long beginTime, long limitspped) {
        //流量控制 100KB/S = 100  * 1024  b/s  = 1024000 b/S =  102 b /ms
        //limitspped KB /s = limitspped * 1024 * 1000 / 1000 = limitspped * 1024  / ms
        long passTime = (System.currentTimeMillis() - beginTime);
        //至少保证时间
       // if (passTime == 0) passTime = 1;
        long pp = passedlen / limitspped - passTime;//(毫秒)//如果passTime=0，按实际所需passedlen/limitspped等待
        if (pp > 0) {
            try {
                Thread.sleep(pp);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //#endregion

    //#region 接收消息


    //#endregion

    //#region 接收文件

    /**
     * 上传文件中间件， 写入文件
     *  接收 FileMessageProtocol
     *  返回 CmdObjectProtocol
     * @param data FileMessageProtocol
     * @param uploadFilePath 上传文件夹（接收端）

     * @param channel
     */
    public static void doTransfer(FileMessageProtocol data, String uploadFilePath, Channel channel) {
        //-----------------------------------
        String fileName = data.getFileName();
        //long fileSize = data.getFileSize();
        //-----------------------------------
        String path = uploadFilePath + File.separator + fileName;
        File file = new File(path);
        //收到文件开始标识，回复pong
        if (data.getTagA().equals("begin")) {
            if (file.exists()) {
                file.delete();
            }
            //告诉对方开始接收数据
            CmdObjectProtocol dto = new CmdObjectProtocol("revicefile",data.getTagB(), "开始接收数据!");
            channel.writeAndFlush(dto);

/*
            RandomAccessFile randomAccessFile = null;
            try {
                randomAccessFile = new RandomAccessFile(file, "rw");
                randomAccessFile.setLength(data.getFileSize());
                randomAccessFile.close();
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
*/

            return;
        }

        //收到文件结束表制，回复pong
        if (data.getTagA().equals("end")) {
            //告诉对方接收数据完毕
            CmdObjectProtocol dto = new CmdObjectProtocol("revicefile",data.getTagB(), "数据接收完毕!");
            channel.writeAndFlush(dto);
            return;
        }
        //读文件的相关代码


        if (data.getTagA().equals("tranfer")) {
            RandomAccessFile randomAccessFile = null;

            try {
                long beginTime =System.currentTimeMillis();
                // 两种情况：
                //* 1.只读时，若文件不存在，则抛出
                //* 2.读写时,若文件不存在  ，则自动创建该文件 ，若创建不成功则抛出
                randomAccessFile = new RandomAccessFile(file, "rw");
                if (file.isFile()) {
                    long start = 0L;
                    byte[] bytes = data.getFileData();
                    long beginPos = data.getBeginPos();
                    long endPos = data.getEndPos();
                    long fileSize = data.getFileSize();


                    start = start + beginPos;
                    randomAccessFile.seek(start);
                    randomAccessFile.write(bytes);
                    randomAccessFile.close();
                    //流量控制 1024/ms = 1M/S
                    flowControl(endPos-beginPos,beginTime,5120);
                    // start = start + byteRead;
                    //告诉对方接收进度
                    CmdObjectProtocol dto = new CmdObjectProtocol("revicefile",data.getTagB(), "接收进度:"+endPos+"|" + ((start + bytes.length) * 100 / fileSize) + "%");
                    channel.writeAndFlush(dto);
                }
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
            finally {
                if (randomAccessFile!=null) {
                    try {
                        randomAccessFile.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    /**
     * 上传文件中间件， 写入文件（custom模式)
     * 接收 CustomMessageProtocol--> FileMessageProtocol
     * 返回 CmdObjectProtocol --> CustomMessageProtocol
     * @param in_cmp CustomMessageProtocol --> FileMessageProtocol
     * @param uploadFilePath 上传文件夹（接收端）
     * @param channel
     */
    public static void doCustomTransfer(CustomMessageProtocol in_cmp, String uploadFilePath, Channel channel, Logger log) throws Exception {
        FileMessageProtocol fmp = (FileMessageProtocol) ByteUtil.BytesToObject(in_cmp.getByteContent());

        //-----------------------------------
        String fileName = fmp.getFileName();
        //long fileSize = data.getFileSize();
        String path = uploadFilePath + File.separator + fileName;
        File file = new File(path);
        //-----------------------------------
        //收到文件开始标识，回复pong
        if (fmp.getTagA().equals("begin")) {

            if (file.exists()) {
                file.delete();
            }
            log.debug("开始接收数据");
            //告诉对方开始接收数据
            CmdObjectProtocol begin_cop = new CmdObjectProtocol("revicefile",fmp.getTagB(), "开始接收数据!");
            CustomMessageProtocol begin_cmp = new CustomMessageProtocol();
            begin_cmp.setContentSystemCmdObject(begin_cop);
            sendObjMsg(begin_cmp, channel);
            return;
        }

        //收到文件结束表制，回复pong
        if (fmp.getTagA().equals("end")) {
            log.debug("数据接收完毕");
            //log.debug("数据接收完毕，大小："+(file.length()/1024/1024)+"M, 耗时： " +(System.currentTimeMillis()-beginTime)/1000 +" 秒 , 平均 "+(file.length()/1024/1024)/(System.currentTimeMillis()-beginTime)/1000+"m/s" );

            //告诉对方接收数据完毕
            CmdObjectProtocol end_cop = new CmdObjectProtocol("revicefile",fmp.getTagB(), "数据接收完毕!");
            CustomMessageProtocol end_cmp = new CustomMessageProtocol();
            end_cmp.setContentSystemCmdObject(end_cop);
            sendObjMsg(end_cmp, channel);
            return;
        }
        //读文件的相关代码
        if (fmp.getTagA().equals("tranfer")) {
            if (fmp.getFileData().length>1024*100)
                throw  new Exception("接收数据包操作限制");
            RandomAccessFile randomAccessFile = null;
            try {
                long beginTime =System.currentTimeMillis();
                // 两种情况：
                //* 1.只读时，若文件不存在，则抛出
                //* 2.读写时,若文件不存在  ，则自动创建该文件 ，若创建不成功则抛出
                randomAccessFile = new RandomAccessFile(file, "rw");
                if (file.isFile()) {
                    long start = 0L;
                    byte[] bytes = fmp.getFileData();
                    long beginPos = fmp.getBeginPos();
                    long endPos = fmp.getEndPos();
                    long fileSize = fmp.getFileSize();


                    start = start + beginPos;
                    randomAccessFile.seek(start);
                    randomAccessFile.write(bytes);
                    randomAccessFile.close();
                    //限流 10240 b/ms = 10240kb/s = 10m/s(实际7m/s)  , 5120 = 5m/s(实际3m/s)
                   // asdf此处服务端控制流量无效，需要重新找地方控制
                   // flowControl(endPos-beginPos,beginTime,104);
                    // start = start + byteRead;
                    //告诉对方接收进度
                    log.debug(fmp.getTagB()+"-》接收进度:"+endPos+"|" + ((start + bytes.length) * 100 / fileSize)+"%");
                    CmdObjectProtocol cop = new CmdObjectProtocol("revicefile",fmp.getTagB(), "接收进度:"+endPos+"|" + ((start + bytes.length) * 100 / fileSize) + "%");
                    CustomMessageProtocol cmp = new CustomMessageProtocol();
                    cmp.setContentSystemCmdObject(cop);
                    sendObjMsg(cmp,channel);
                }
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
            finally {
                if (randomAccessFile!=null) {
                    try {
                        randomAccessFile.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

            }
        }
    }



    //#endregion
}
